package com.bml.architect.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * kafka生产者
 */
public class KafkaProducer {

    private final static String topic = "openapi";

    public static void main(String[] args) throws ExecutionException, InterruptedException, IOException {

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.108.247.235:9092");
        //acks=0：表示producer发送消息到broker过后，不需要等待broker回复任何消息就可以发送下一条消息，性能最高，但是容易丢消息；
        //acks=1:表示至少要等待leader接受到消息并且成功写入leader的本地log文件，而不需要等待follower是否成功写入，就可以发送下一条消息
        //这种情况，如果follower没有成功备份数据，而此时leader又挂掉了，那么消息会丢失
        //acks=-1或者all:表示需要等待min.insync.replicas（默认配置为1，推荐配置为2）这个参数配置的副本数都是成功写入本地log才返回成功
        //这种策略会保证 只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        //发送失败会重试，默认重试的间隔是100ms，重试会保证消息的成功发送，但是也有可能出现网络抖动出现消息的重复发送，这种情况下，就需要消费者
//        做好消息的消费的幂等性处理
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        //消费发送失败的重试时间间隔
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        //设置发送消息的缓冲区，如果设置了缓冲区，消息可先发送到缓冲区，然后再从缓冲区读取消息，这样可以提供消息的发送性能，默认是33554432，即32M
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        //kafka本地线程会从缓冲区读取数据，批量发送到broker，设置批量发送消息的大小，默认是16384，即16k，就是说一个batch满了16k就会发送出去
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        //默认值是0，意思就是消息必须立即被发送，但这样会影响性能, 一般设置10毫秒左右，就是说这个消息发送完后会进入本地的一个batch
        //如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去,如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        //把发送的key从字符串序列化为字节数组
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<String, String>(properties);

        for (int i = 0; i < 5; i++) {
            Order order = new Order(i, 100 + i, 1, 1000.00);
            //指定分区发送
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, 0, JSON.toJSONString(order.getOrderId()), JSON.toJSONString(order));
            //不指定分区发送,具体发送的分区计算公式：hash(key)%partitionNum
            // ProducerRecord<String,String> record = new ProducerRecord<String,String>(topic,JSON.toJSONString(order.getOrderId()),JSON.toJSONString(order));

            //等待消息发送成功的同步阻塞方法
//            RecordMetadata recordMetadata = producer.send(record).get();
//            System.out.println("同步方式发送消息结果：" + "topic-" + recordMetadata.topic() + "|partition-"
//                    + recordMetadata.partition() + "|offset-" + recordMetadata.offset());
            //异步发送,在发送成功以后会通过回调函数来通知
             producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {

                        System.out.println("发送出现异常:" + exception.getMessage());
                    } else {
                        System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                                + metadata.partition() + "|offset-" + metadata.offset());
                    }

                }
            });

        }
        producer.close();
        System.in.read();




    }
}
